package com.jwd.avmanager;


import com.jwd.avmanager.constant.Cmd;
import com.jwd.avmanager.parse.ProtocolParser;
import com.jwd.avmanager.repository.BDMybais;
import com.jwd.avmanager.server.*;
import com.jwd.avmanager.util.DebugLogger;
import com.jwd.avmanager.util.HttpRequestUtils;
import com.jwd.avmanager.util.ResponseFactory;
import com.jwd.avmanager.util.ResponseSendUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.bytes.ByteArrayEncoder;
import io.netty.handler.codec.http.HttpContentCompressor;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ManagerBootStrap {
    static Logger logger = LoggerFactory.getLogger(ManagerBootStrap.class);
    //    Logger logger = Logger.getLogger("zhifayi");
    public static final int TCP_PORT = 3503;//执法仪

    //创建tcp连接池
    private EventLoopGroup tcpBossGroup = new NioEventLoopGroup();
    private EventLoopGroup tcpWorkGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);

    private ExecutorService service = Executors.newFixedThreadPool(2);//线程池启动，根据实际启动的服务调整
    private Timer timer = new Timer();
    public static void main(String[] args) {

        ManagerBootStrap managerBootStrap = new ManagerBootStrap();
        managerBootStrap.startServer();

//        Login getuser = loginMapper.getlogin(21);
//        DebugLogger.debug(getuser.getUserName());
    }

    /**
     * 启动tcp服务程序
     */
    class TcpServerTask implements Runnable {
        @Override
        public void run() {
            DebugLogger.info("启动");
            startTcpServer();
        }
    }

    /**
     * 刷设备用户状态
     */
    class BrushDeviceUserStatus implements Runnable {
        @Override
        public void run() {
            DebugLogger.info("启动MQL");
            BDMybais instance = BDMybais.getInstance();
            instance.BrushDeviceUerStatus();

        }
    }
    /**
     * 开启rsr 流检查  5分钟一次
     */
    class SRSServerTask implements Runnable {
        @Override
        public void run() {
            DebugLogger.info("启动流检查");
            // 创建一个定时任务
            TimerTask task = new TimerTask() {
                @Override
                public void run() {
                    DebugLogger.info("SRS检查");
                    String  url = "http://47.119.165.185:1985/api/v1/streams?start=0";
                    List<String> list =null;
                    try {
                        list = HttpRequestUtils.SRSHttp(url);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    if( list == null ||list.size() == 0 ){

                    } else {
                        for (String id : list) {
//                            DebugLogger.debug(id +" 80");
                            ChannelHandlerContext socketClient = ClientManager.getInstance().getSocketClient(String.valueOf(id));
                            byte[] response = ResponseFactory.getResponse(Cmd.CMD_OUT_STOP_RTMP, (short) 0, Integer.parseInt(id));
                            DebugLogger.info("淘汰流 devid:->"+Integer.parseInt(id));
                            if(socketClient!=null){
                                ResponseSendUtil.writeAndFlush(response, socketClient);
                            }
                        }
                    }
                }
            };

            // 在 3 分钟后执行定时任务，并在每次执行后等待 20秒
            timer.schedule(task, 3*60*1000, 1*60*1000);
        }
    }

    /**
     * 开启服务
     */
    public void startServer() {
        new Thread(new WebsocketServer(),"WebsocketTh").start();
        service.execute(new TcpServerTask());
        service.execute(new SRSServerTask());
        //TODO 刷新设备在线状态-刷新用户在线状态
//        service.execute(new BrushDeviceUserStatus());
    }

    /**
     * 配置tcpserver并启动
     */
    private void startTcpServer() {
        try {
            //创建服务端启动对象，配置参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(tcpBossGroup, tcpWorkGroup)
                    .channel(NioServerSocketChannel.class)//设置通道类型
                    .option(ChannelOption.SO_BACKLOG, 65535)//设置线程队列等待连接个数
                    .option(ChannelOption.SO_RCVBUF, 1024*20)
                    .option(ChannelOption.RCVBUF_ALLOCATOR,new FixedRecvByteBufAllocator(1024*20))
                    .childOption(ChannelOption.SO_KEEPALIVE, true)//连接保活
                    .childOption(ChannelOption.SINGLE_EVENTEXECUTOR_PER_GROUP, false)
                    .childOption(ChannelOption.TCP_NODELAY,true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {//创建一个通道初始化对象
                        //给pipeline设置处理器
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            //进来数据封装
                            ch.pipeline().addLast(new CustomMessageDecoder());
                            //出去数据的封装
                            ch.pipeline().addLast(new ByteArrayEncoder());
                            //超时设置
                            ch.pipeline().addLast(new IdleStateHandler(30, 30, 30, TimeUnit.SECONDS));
                            //业务处理
                            ch.pipeline().addLast(new SignalServerHandler());
                        }
                    });
            DebugLogger.info("tcp server is ready boot");
            ChannelFuture channelFuture = bootstrap.bind(TCP_PORT).sync();
            channelFuture.addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    DebugLogger.info("tcp server run success");
                    ClientManager.getInstance();//初始化一下
                }
            });
            //对关闭通道进行监听
            ChannelFuture closeFuture = channelFuture.channel().closeFuture().await();
            closeFuture.addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    DebugLogger.info("tcp server 关闭完成");
                }
            });
            //释放工作
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            try {
                tcpBossGroup.shutdownGracefully().sync();
                tcpWorkGroup.shutdownGracefully().sync();
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        }
    }

    /**
     * 启动http服务
     */
    private void startHttpServer(){

        JedisClient.initJedis();
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors());
        ServerBootstrap bootstrap = new ServerBootstrap();
        try {
            bootstrap.group(bossGroup, workGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel channel) {
                            channel.pipeline().addLast("HttpServerCodec", new HttpServerCodec());
                            channel.pipeline().addLast("compressor", new HttpContentCompressor());
                            channel.pipeline().addLast("aggregator", new HttpObjectAggregator(512 * 1024));
                            channel.pipeline().addLast("HTTPServerHandler", new HttpRequestHandler());
                        }
                    });
            ChannelFuture channelFuture = bootstrap.bind(10800).syncUninterruptibly();
            channelFuture.addListener((ChannelFutureListener) channelFuture1 -> {
                boolean success = channelFuture1.isSuccess();
                System.out.println("启动结果:" + success);
            });
            channelFuture.channel().closeFuture().syncUninterruptibly();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                bossGroup.shutdownGracefully().sync();
                workGroup.shutdownGracefully().sync();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

        }
    }

}
